package org.lemon.transport.mqtt.server.handler;

import com.fasterxml.jackson.databind.JsonNode;
import com.google.gson.JsonParseException;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.handler.codec.mqtt.*;
import io.netty.handler.ssl.SslHandler;
import io.netty.util.CharsetUtil;
import io.netty.util.ReferenceCountUtil;
import io.netty.util.concurrent.Future;
import io.netty.util.concurrent.GenericFutureListener;
import lombok.extern.slf4j.Slf4j;
import org.lemon.transport.context.adaptor.AdaptorException;
import org.lemon.transport.context.auth.TransportDeviceInfo;
import org.lemon.transport.context.auth.ValidateDeviceCredentialsResponse;
import org.lemon.transport.context.auth.SessionInfoCreator;
import org.lemon.transport.context.data.device.*;
import org.lemon.transport.context.service.*;
import org.lemon.transport.mqtt.server.session.DeviceSessionContext;
import org.lemon.transport.mqtt.server.model.MqttTopicMatcher;
import org.server.transport.gen.SessionProto;
import org.server.transport.gen.TransportProto;
import org.lemon.transport.context.model.SessionMetaData;
import org.lemon.transport.context.data.TransportPayloadType;
import org.lemon.transport.mqtt.server.BootNettyMqttServer;
import org.lemon.transport.mqtt.server.adaptors.MqttTransportAdaptor;
import org.lemon.transport.mqtt.server.adaptors.MqttTransportAdaptorFactory;
import org.lemon.transport.mqtt.server.model.TopicType;
import org.lemon.transport.mqtt.server.service.MqttTransportContext;
import org.lemon.transport.context.session.SessionMsgListener;
import org.lemon.transport.context.utils.SslUtil;

import javax.net.ssl.SSLPeerUnverifiedException;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.security.cert.Certificate;
import java.security.cert.X509Certificate;
import java.util.ArrayList;
import java.util.List;
import java.util.Optional;
import java.util.UUID;
import java.util.concurrent.Callable;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

import static io.netty.handler.codec.mqtt.MqttConnectReturnCode.CONNECTION_ACCEPTED;
import static io.netty.handler.codec.mqtt.MqttConnectReturnCode.CONNECTION_REFUSED_NOT_AUTHORIZED;
import static io.netty.handler.codec.mqtt.MqttMessageType.*;
import static io.netty.handler.codec.mqtt.MqttQoS.*;

/**
 *
 */
@Slf4j
public class MqttTransportHandler extends ChannelInboundHandlerAdapter implements GenericFutureListener<Future<? super Void>>, SessionMsgListener {

    /**
     * 固件包请求
     */
    private static final Pattern FW_REQUEST_PATTERN = Pattern.compile(MqttTopics.DEVICE_FIRMWARE_REQUEST_TOPIC_PATTERN);
    /**
     * 软件包请求
     */
    private static final Pattern SW_REQUEST_PATTERN = Pattern.compile(MqttTopics.DEVICE_SOFTWARE_REQUEST_TOPIC_PATTERN);

    private final UUID sessionId;
    private final MqttTransportContext context;
    private final TransportService transportService;
    private final SchedulerComponent scheduler;
    private final ConcurrentHashMap<MqttTopicMatcher, Integer> mqttQosMap;

    private final DeviceSessionContext deviceSessionContext;
    volatile InetSocketAddress address;
    volatile GatewaySessionHandler gatewaySessionHandler;
    private final boolean sslEnabled;
    private final SslHandler sslHandler;

    /**
     * 属性 topic 类型
     */
    private TopicType attrSubTopicType;
    /**
     * rpc topic 类型
     */
    private TopicType rpcSubTopicType;
    /**
     * 属性 topic 类型
     */
    private TopicType attrReqTopicType;
    /**
     * to server rpc topic 类型
     */
    private TopicType toServerRpcSubTopicType;

    public MqttTransportHandler(MqttTransportContext context, boolean sslEnabled, SslHandler sslHandler) {
        this.sessionId = UUID.randomUUID();
        this.context = context;
        this.transportService = context.getTransportService();
        this.scheduler = context.getScheduler();
        this.mqttQosMap = new ConcurrentHashMap<>();
        this.deviceSessionContext = new DeviceSessionContext(sessionId, mqttQosMap, context);
        this.sslEnabled = sslEnabled;
        this.sslHandler = sslHandler;
    }

    @Override
    public void channelRegistered(ChannelHandlerContext ctx) throws Exception {
        super.channelRegistered(ctx);
        context.channelRegistered();
    }

    @Override
    public void channelUnregistered(ChannelHandlerContext ctx) throws Exception {
        super.channelUnregistered(ctx);
        context.channelUnregistered();
    }

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {

        log.trace("[{}] Processing msg: {}", sessionId, msg);
        if (this.address == null) {
            this.address = getAddress(ctx);
        }
        try {
            if (msg instanceof MqttMessage) {
                MqttMessage message = (MqttMessage) msg;
                if (message.decoderResult().isSuccess()) {
                    processMqttMsg(ctx, message);
                } else {
                    log.error("[{}] Message decoding failed: {}", sessionId, message.decoderResult().cause().getMessage());
                    ctx.close();
                }
            } else {
                log.debug("[{}] Received non mqtt message: {}", sessionId, msg.getClass().getSimpleName());
                ctx.close();
            }
        } finally {
            //最后一个channelHandler 需要手动release消息(释放内存)
            ReferenceCountUtil.safeRelease(msg);
        }
    }

    @Override
    public void operationComplete(Future<? super Void> future) throws Exception {
        log.trace("[{}] Channel closed!", sessionId);
    }

    void processMqttMsg(ChannelHandlerContext ctx, MqttMessage msg) {

        if (msg.fixedHeader() == null) {
            log.info("[{}:{}] Invalid message received", address.getHostName(), address.getPort());
            ctx.close();
            return;
        }
        if (CONNECT.equals(msg.fixedHeader().messageType())) {
            processConnect(ctx, (MqttConnectMessage) msg);
        } else if (deviceSessionContext.isProvisionOnly()) {
            processProvisionSessionMsg(ctx, msg);
        } else {
            enqueueRegularSessionMsg(ctx, msg);
        }
    }

    private void processProvisionSessionMsg(ChannelHandlerContext ctx, MqttMessage msg) {

        switch (msg.fixedHeader().messageType()) {
            case PUBLISH:
                MqttPublishMessage mqttPubMsg = (MqttPublishMessage) msg;
                String topicName = mqttPubMsg.variableHeader().topicName();
                int msgId = mqttPubMsg.variableHeader().packetId();
                try {
                    if (topicName.equals(MqttTopics.DEVICE_PROVISION_REQUEST_TOPIC)) {
                        try {
                            TransportProto.ProvisionDeviceRequestMsg provisionRequestMsg = deviceSessionContext.getAdaptor(attrSubTopicType).convertToProvisionRequestMsg(deviceSessionContext, mqttPubMsg);
                            transportService.process(provisionRequestMsg, new DeviceProvisionCallback(ctx, msgId, msg.fixedHeader(), provisionRequestMsg));
                            log.trace("[{}][{}] Processing provision publish msg [{}][{}]!", sessionId, deviceSessionContext.getDeviceId(), topicName, msgId);
                        } catch (Exception e) {
                            if (e instanceof JsonParseException || (e.getCause() != null && e.getCause() instanceof JsonParseException)) {
                                TransportProto.ProvisionDeviceRequestMsg provisionRequestMsg = MqttTransportAdaptorFactory.getProtoMqttAdaptor().convertToProvisionRequestMsg(deviceSessionContext, mqttPubMsg);
                                transportService.process(provisionRequestMsg, new DeviceProvisionCallback(ctx, msgId, msg.fixedHeader(), provisionRequestMsg));
                                deviceSessionContext.setProvisionPayloadType(TransportPayloadType.PROTOBUF);
                                log.trace("[{}][{}] Processing provision publish msg [{}][{}]!", sessionId, deviceSessionContext.getDeviceId(), topicName, msgId);
                            } else {
                                throw e;
                            }
                        }
                    } else {
                        log.debug("[{}] Unsupported topic for provisioning requests: {}!", sessionId, topicName);
                        ctx.close();
                    }
                } catch (RuntimeException | AdaptorException e) {
                    log.warn("[{}] Failed to process publish msg [{}][{}]", sessionId, topicName, msgId, e);
                    ctx.close();
                }
            case PINGREQ:
                ctx.writeAndFlush(new MqttMessage(new MqttFixedHeader(PINGRESP, false, AT_MOST_ONCE, false, 0)));
                break;
            case DISCONNECT:
                ctx.close();
                break;
        }
    }

    /**
     * session 消息入队列处理
     *
     * @param ctx
     * @param msg
     */
    private void enqueueRegularSessionMsg(ChannelHandlerContext ctx, MqttMessage msg) {

        this.deviceSessionContext.addMessageToQueue(msg);
        this.deviceSessionContext.tryProcessQueueMessage(mqttMsg -> processRegularSessionMsg(ctx, mqttMsg));
    }

    /**
     * 定期处理回话消息
     *
     * @param ctx
     * @param msg
     */
    private void processRegularSessionMsg(ChannelHandlerContext ctx, MqttMessage msg) {
        switch (msg.fixedHeader().messageType()) {
            case PUBLISH:       //发布消息
                processPublish(ctx, (MqttPublishMessage) msg);
                break;
            case SUBSCRIBE:     //客户端主题订阅
                processSubscribe(ctx, (MqttSubscribeMessage) msg);
                break;
            case UNSUBSCRIBE:   //取消订阅
                processUnsubscribe(ctx, (MqttUnsubscribeMessage) msg);
                break;
            case PINGREQ:       //ping 请求
                //ping 响应固定头
                MqttFixedHeader fixedHeader = new MqttFixedHeader(PINGRESP, false, AT_MOST_ONCE, false, 0);
                ctx.writeAndFlush(new MqttMessage(fixedHeader));
                break;
            case DISCONNECT:    //断开连接
                ctx.close();
                break;
            case PUBACK:         //发布应答 qos = 2 才有
                break;
        }
    }

    /**
     * 处理连接事件
     *
     * @param ctx
     * @param connectMsg
     */
    void processConnect(ChannelHandlerContext ctx, MqttConnectMessage connectMsg) {

        String userName = connectMsg.payload().userName();
        String clientId = connectMsg.payload().clientIdentifier();
        log.debug("[{}][{}] Processing connect msg for client: {}!", address, sessionId, clientId);
        if (userName.equals("provision") || clientId.equals("provision")) {
            deviceSessionContext.setChannel(ctx);
            deviceSessionContext.setProvisionOnly(true);
            log.info("connection success: {} session-id[{}]", deviceSessionContext.getClientId(), this.sessionId);
            ctx.writeAndFlush(createMqttConnAckMsg(CONNECTION_ACCEPTED, connectMsg));
        } else {
            X509Certificate cert;
            if (sslEnabled && sslHandler != null && (cert = getX509Certificate()) != null) {
                processX509CertConnect(ctx, cert, connectMsg);
            } else {
                processBasicCredConnect(ctx, connectMsg);
            }
        }
    }

    /**
     * 处理发布消息
     *
     * @param ctx
     * @param mqttMsg
     */
    private void processPublish(ChannelHandlerContext ctx, MqttPublishMessage mqttMsg) {

        if (!checkConnected(ctx, mqttMsg)) {
            return;
        }
        String topicName = mqttMsg.variableHeader().topicName();
        int msgId = mqttMsg.variableHeader().packetId();
        log.trace("[{}] Processing publish msg [{}][{}]!", sessionId, topicName, msgId);

        if (topicName.startsWith(MqttTopics.BASE_GATEWAY_API_TOPIC)) {
            //来自设备网关发布的消息
            if (gatewaySessionHandler != null) {
                processGatewayPublishMessage(ctx, mqttMsg, topicName, msgId);
                transportService.reportActivity(deviceSessionContext.getSessionInfo());
            }
        } else {
            //来自设备发布的消息
            processDevicePublishMessage(ctx, mqttMsg, topicName, msgId);
        }
    }

    private void processGatewayPublishMessage(ChannelHandlerContext ctx, MqttPublishMessage mqttMsg, String topicName, int msgId) {

        try {
            switch (topicName) {
                case MqttTopics.GATEWAY_TELEMETRY_TOPIC:
                    gatewaySessionHandler.onDeviceTelemetry(mqttMsg);
                    break;
                case MqttTopics.GATEWAY_CLAIM_TOPIC:
                    gatewaySessionHandler.onDeviceClaim(mqttMsg);
                    break;
                case MqttTopics.GATEWAY_ATTRIBUTES_TOPIC:
                    gatewaySessionHandler.onDeviceAttributes(mqttMsg);
                    break;
                case MqttTopics.GATEWAY_ATTRIBUTES_REQUEST_TOPIC:
                    gatewaySessionHandler.onDeviceAttributesRequest(mqttMsg);
                    break;
                case MqttTopics.GATEWAY_RPC_TOPIC:
                    gatewaySessionHandler.onDeviceRpc(mqttMsg);
                    break;
                case MqttTopics.GATEWAY_CONNECT_TOPIC:
                    gatewaySessionHandler.onDeviceConnect(mqttMsg);
                    break;
                case MqttTopics.GATEWAY_DISCONNECT_TOPIC:
                    gatewaySessionHandler.onDeviceDisconnect(mqttMsg);
                    break;
                default:
                    publishAck(ctx, msgId, mqttMsg.fixedHeader());
            }
        } catch (RuntimeException e) {
            log.warn("[{}] Failed to process publish msg [{}][{}]", sessionId, topicName, msgId, e);
            ctx.close();
        } catch (AdaptorException e) {
            log.debug("[{}] Failed to process publish msg [{}][{}]", sessionId, topicName, msgId, e);
            sendAckOrCloseSession(ctx, topicName, msgId, mqttMsg.fixedHeader());
        }
    }

    /**
     * 设备发布的消息处理
     *
     * @param ctx
     * @param mqttMsg
     * @param topicName
     * @param msgId
     */
    private void processDevicePublishMessage(ChannelHandlerContext ctx, MqttPublishMessage mqttMsg, String topicName, int msgId) {

        try {
            MqttTransportAdaptor payloadAdaptor = this.deviceSessionContext.getPayloadAdaptor();
            Matcher matcher;
            if (deviceSessionContext.isDeviceAttributesTopic(topicName)) {
                //处理设备属性数据的topic
                TransportProto.PostAttributeMsg postAttributeMsg = payloadAdaptor.convertToPostAttributes(deviceSessionContext, mqttMsg);
                transportService.process(this.deviceSessionContext.getSessionInfo(), postAttributeMsg, getPubAckCallback(ctx, msgId, mqttMsg.fixedHeader(), postAttributeMsg));
            } else if (deviceSessionContext.isDeviceTelemetryTopic(topicName)) {
                //处理设备遥测数据的topic
                TransportProto.PostTelemetryMsg postTelemetryMsg = payloadAdaptor.convertToPostTelemetry(deviceSessionContext, mqttMsg);
                transportService.process(this.deviceSessionContext.getSessionInfo(), postTelemetryMsg, getPubAckCallback(ctx, msgId, mqttMsg.fixedHeader(), postTelemetryMsg));
            } else if (topicName.startsWith(MqttTopics.DEVICE_ATTRIBUTES_REQUEST_TOPIC_PREFIX)) {
                //设备属性请求
                attrReqTopicType = TopicType.V1;
            } else if (topicName.startsWith(MqttTopics.DEVICE_RPC_REQUESTS_TOPIC)) {
                //设备rpc请求
                toServerRpcSubTopicType = TopicType.V1;
            } else if (topicName.startsWith(MqttTopics.DEVICE_RPC_RESPONSE_TOPIC)) {
                //设备rpc响应
            } else if (topicName.equals(MqttTopics.DEVICE_CLAIM_TOPIC)) {
                //设备分配
            } else if ((matcher = FW_REQUEST_PATTERN.matcher(topicName)).find()) {
                //todo 固件请求
            } else if ((matcher = SW_REQUEST_PATTERN.matcher(topicName)).find()) {
                //todo 软件请求
            } else if (topicName.equals(MqttTopics.DEVICE_TELEMETRY_SHORT_TOPIC)) {
                //v2版设备遥测数据
                TransportProto.PostTelemetryMsg postTelemetryMsg = payloadAdaptor.convertToPostTelemetry(deviceSessionContext, mqttMsg);
                transportService.process(this.deviceSessionContext.getSessionInfo(), postTelemetryMsg, getPubAckCallback(ctx, msgId, mqttMsg.fixedHeader(), postTelemetryMsg));
            } else if (topicName.equals(MqttTopics.DEVICE_TELEMETRY_SHORT_JSON_TOPIC)) {
                TransportProto.PostTelemetryMsg postTelemetryMsg = MqttTransportAdaptorFactory.getJsonMqttAdaptor().convertToPostTelemetry(deviceSessionContext, mqttMsg);
                transportService.process(this.deviceSessionContext.getSessionInfo(), postTelemetryMsg, getPubAckCallback(ctx, msgId, mqttMsg.fixedHeader(), postTelemetryMsg));
            } else if (topicName.equals(MqttTopics.DEVICE_TELEMETRY_SHORT_PROTO_TOPIC)) {
                TransportProto.PostTelemetryMsg postTelemetryMsg = MqttTransportAdaptorFactory.getProtoMqttAdaptor().convertToPostTelemetry(deviceSessionContext, mqttMsg);
                transportService.process(this.deviceSessionContext.getSessionInfo(), postTelemetryMsg, getPubAckCallback(ctx, msgId, mqttMsg.fixedHeader(), postTelemetryMsg));
            } else if (topicName.equals(MqttTopics.DEVICE_ATTRIBUTES_SHORT_TOPIC)) {
                //v2版设备属性数据
                TransportProto.PostAttributeMsg postAttributeMsg = payloadAdaptor.convertToPostAttributes(deviceSessionContext, mqttMsg);
                transportService.process(this.deviceSessionContext.getSessionInfo(), postAttributeMsg, getPubAckCallback(ctx, msgId, mqttMsg.fixedHeader(), postAttributeMsg));
            } else if (topicName.equals(MqttTopics.DEVICE_ATTRIBUTES_SHORT_JSON_TOPIC)) {
                TransportProto.PostAttributeMsg postAttributeMsg = MqttTransportAdaptorFactory.getJsonMqttAdaptor().convertToPostAttributes(deviceSessionContext, mqttMsg);
                transportService.process(this.deviceSessionContext.getSessionInfo(), postAttributeMsg, getPubAckCallback(ctx, msgId, mqttMsg.fixedHeader(), postAttributeMsg));
            } else if (topicName.equals(MqttTopics.DEVICE_ATTRIBUTES_SHORT_PROTO_TOPIC)) {
                TransportProto.PostAttributeMsg postAttributeMsg = MqttTransportAdaptorFactory.getProtoMqttAdaptor().convertToPostAttributes(deviceSessionContext, mqttMsg);
                transportService.process(this.deviceSessionContext.getSessionInfo(), postAttributeMsg, getPubAckCallback(ctx, msgId, mqttMsg.fixedHeader(), postAttributeMsg));
            } else if (topicName.startsWith(MqttTopics.DEVICE_RPC_RESPONSE_SHORT_JSON_TOPIC)) {
                //todo 设备rpc响应
            } else if (topicName.startsWith(MqttTopics.DEVICE_RPC_RESPONSE_SHORT_PROTO_TOPIC)) {

            } else if (topicName.startsWith(MqttTopics.DEVICE_RPC_RESPONSE_SHORT_TOPIC)) {

            } else if (topicName.startsWith(MqttTopics.DEVICE_RPC_REQUESTS_SHORT_JSON_TOPIC)) {
                //todo 设备rpc请求
                toServerRpcSubTopicType = TopicType.V2_JSON;
            } else if (topicName.startsWith(MqttTopics.DEVICE_RPC_REQUESTS_SHORT_PROTO_TOPIC)) {

                toServerRpcSubTopicType = TopicType.V2_PROTO;
            } else if (topicName.startsWith(MqttTopics.DEVICE_RPC_REQUESTS_SHORT_TOPIC)) {

                toServerRpcSubTopicType = TopicType.V2;
            } else if (topicName.startsWith(MqttTopics.DEVICE_ATTRIBUTES_REQUEST_SHORT_JSON_TOPIC_PREFIX)) {
                //todo 设备属性请求
                attrReqTopicType = TopicType.V2_JSON;
            } else if (topicName.startsWith(MqttTopics.DEVICE_ATTRIBUTES_REQUEST_SHORT_PROTO_TOPIC_PREFIX)) {

                attrReqTopicType = TopicType.V2_PROTO;
            } else if (topicName.startsWith(MqttTopics.DEVICE_ATTRIBUTES_REQUEST_SHORT_TOPIC_PREFIX)) {

                attrReqTopicType = TopicType.V2;
            } else {
                publishAck(ctx, msgId, mqttMsg.fixedHeader());
            }
        } catch (Exception e) {
            log.error("[{}] Fail to process publish message [{}] [{}]", sessionId, topicName, msgId, e);
            sendAckOrCloseSession(ctx, topicName, msgId, mqttMsg.fixedHeader());
        }
    }

    private void sendAckOrCloseSession(ChannelHandlerContext ctx, String topicName, int msgId, MqttFixedHeader mqttFixedHeader) {
        if (deviceSessionContext.isSendAckOnValidationException() && msgId > 0) {
            log.debug("[{}] Send pub ack on invalid publish msg [{}][{}]", sessionId, topicName, msgId);
            //ctx.writeAndFlush(createMqttPubAckMsg(msgId));
            publishAck(ctx, msgId, mqttFixedHeader);
        } else {
            log.info("[{}] Closing current session due to invalid publish msg [{}][{}]", sessionId, topicName, msgId);
            ctx.close();
        }
    }

    /**
     * 消息发布应答
     *
     * @param ctx
     * @param requestId
     * @param mqttFixedHeader
     */
    public static void publishAck(ChannelHandlerContext ctx, int requestId, MqttFixedHeader mqttFixedHeader) {

        MqttQoS qos = mqttFixedHeader.qosLevel();
        switch (qos) {
            case AT_MOST_ONCE:        //	至多一次
                break;
            case AT_LEAST_ONCE:        //	至少一次

                MqttFixedHeader mqttFixedHeaderBack = new MqttFixedHeader(MqttMessageType.PUBACK, mqttFixedHeader.isDup(), MqttQoS.AT_MOST_ONCE, mqttFixedHeader.isRetain(), 0x02);
                MqttMessageIdVariableHeader mqttMessageIdVariableHeaderBack = MqttMessageIdVariableHeader.from(requestId);
                MqttPubAckMessage pubAck = new MqttPubAckMessage(mqttFixedHeaderBack, mqttMessageIdVariableHeaderBack);
                ctx.writeAndFlush(pubAck);
                break;
            case EXACTLY_ONCE:        //	刚好一次

                MqttFixedHeader mqttFixedHeaderBack2 = new MqttFixedHeader(MqttMessageType.PUBREC, false, MqttQoS.AT_LEAST_ONCE, false, 0x02);
                MqttMessageIdVariableHeader mqttMessageIdVariableHeaderBack2 = MqttMessageIdVariableHeader.from(requestId);
                MqttMessage mqttMessageBack = new MqttMessage(mqttFixedHeaderBack2, mqttMessageIdVariableHeaderBack2);
                ctx.writeAndFlush(mqttMessageBack);
                break;
            default:
                break;
        }
    }

    /**
     * 创建发布应答消息
     *
     * @param requestId
     * @return
     */
    public static MqttPubAckMessage createMqttPubAckMsg(int requestId) {

        MqttFixedHeader mqttFixedHeader =
                new MqttFixedHeader(PUBACK, false, AT_MOST_ONCE, false, 0);
        MqttMessageIdVariableHeader mqttMsgIdVariableHeader =
                MqttMessageIdVariableHeader.from(requestId);
        return new MqttPubAckMessage(mqttFixedHeader, mqttMsgIdVariableHeader);
    }

    /**
     * 处理取消订阅
     *
     * @param ctx
     * @param mqttMsg
     */
    private void processUnsubscribe(ChannelHandlerContext ctx, MqttUnsubscribeMessage mqttMsg) {
        if (!checkConnected(ctx, mqttMsg)) {
            return;
        }
        boolean activityReported = false;
        log.trace("[{}] Processing unsubscription [{}]!", sessionId, mqttMsg.variableHeader().messageId());
        for (String topicName : mqttMsg.payload().topics()) {
            mqttQosMap.remove(new MqttTopicMatcher(topicName));
            try {
                switch (topicName) {
                    case MqttTopics.DEVICE_ATTRIBUTES_TOPIC:
                    case MqttTopics.DEVICE_ATTRIBUTES_SHORT_TOPIC:
                    case MqttTopics.DEVICE_ATTRIBUTES_SHORT_PROTO_TOPIC:
                    case MqttTopics.DEVICE_ATTRIBUTES_SHORT_JSON_TOPIC:
                        //TODO 设备属性消息取消订阅
                        activityReported = true;
                        break;
                    case MqttTopics.DEVICE_RPC_REQUESTS_SUB_TOPIC:
                    case MqttTopics.DEVICE_RPC_REQUESTS_SUB_SHORT_TOPIC:
                    case MqttTopics.DEVICE_RPC_REQUESTS_SUB_SHORT_JSON_TOPIC:
                    case MqttTopics.DEVICE_RPC_REQUESTS_SUB_SHORT_PROTO_TOPIC:
                        //TODO 设备rpc消息取消订阅
                        activityReported = true;
                        break;
                    case MqttTopics.DEVICE_RPC_RESPONSE_SUB_TOPIC:
                    case MqttTopics.DEVICE_RPC_RESPONSE_SUB_SHORT_TOPIC:
                    case MqttTopics.DEVICE_RPC_RESPONSE_SUB_SHORT_JSON_TOPIC:
                    case MqttTopics.DEVICE_RPC_RESPONSE_SUB_SHORT_PROTO_TOPIC:
                    case MqttTopics.DEVICE_ATTRIBUTES_RESPONSES_TOPIC:
                    case MqttTopics.DEVICE_ATTRIBUTES_RESPONSES_SHORT_TOPIC:
                    case MqttTopics.DEVICE_ATTRIBUTES_RESPONSES_SHORT_JSON_TOPIC:
                    case MqttTopics.DEVICE_ATTRIBUTES_RESPONSES_SHORT_PROTO_TOPIC:
                    case MqttTopics.GATEWAY_ATTRIBUTES_TOPIC:
                    case MqttTopics.GATEWAY_RPC_TOPIC:
                    case MqttTopics.GATEWAY_ATTRIBUTES_RESPONSE_TOPIC:
                    case MqttTopics.DEVICE_PROVISION_RESPONSE_TOPIC:
                    case MqttTopics.DEVICE_FIRMWARE_RESPONSES_TOPIC:
                    case MqttTopics.DEVICE_FIRMWARE_ERROR_TOPIC:
                    case MqttTopics.DEVICE_SOFTWARE_RESPONSES_TOPIC:
                    case MqttTopics.DEVICE_SOFTWARE_ERROR_TOPIC: {
                        activityReported = true;
                        break;
                    }
                }
            } catch (Exception e) {
                log.debug("[{}] Failed to process unsubscription [{}] to [{}]", sessionId, mqttMsg.variableHeader().messageId(), topicName);
            }
        }
        if (!activityReported) {
            //TODO 事件上报
        }
        //取消订阅应答
        MqttFixedHeader fixedHeader = new MqttFixedHeader(UNSUBACK, false, AT_MOST_ONCE, false, 0);
        MqttMessageIdVariableHeader mqttMessageIdVariableHeader = MqttMessageIdVariableHeader.from(mqttMsg.variableHeader().messageId());
        ctx.writeAndFlush(new MqttMessage(fixedHeader, mqttMessageIdVariableHeader));
    }

    /**
     * 处理客户端的主题订阅消息
     * 客户端向服务端发送SUBSCRIBE报文用于创建一个或多个订阅，每个订阅注册客户端关心的一个或多个主题。
     * 为了将应用消息转发给与那些订阅匹配的主题，服务端发送PUBLISH报文给客户端。
     * SUBSCRIBE报文也（为每个订阅）指定了最大的QoS等级，服务端根据这个发送应用消息给客户端
     *
     * @param ctx
     * @param mqttMsg
     */
    private void processSubscribe(ChannelHandlerContext ctx, MqttSubscribeMessage mqttMsg) {

        if (!checkConnected(ctx, mqttMsg)) {
            return;
        }
        log.trace("[{}] Processing subscription [{}]!", sessionId, mqttMsg.variableHeader().messageId());

        List<MqttTopicSubscription> topicSubscriptions = mqttMsg.payload().topicSubscriptions();
        List<Integer> grantedQoSList = new ArrayList<>(topicSubscriptions.size());

        boolean activityReported = false;
        for (MqttTopicSubscription subscription : topicSubscriptions) {
            String topicName = subscription.topicName();
            MqttQoS reqQoS = subscription.qualityOfService();
            log.info("[{}] Subscription: topic[{}] qos[{}]!", sessionId, topicName, reqQoS);
            try {
                switch (topicName) {
                    case MqttTopics.DEVICE_ATTRIBUTES_TOPIC:
                        processAttributesSubscribe(grantedQoSList, topicName, reqQoS, TopicType.V1);
                        activityReported = true;
                        break;
                    case MqttTopics.DEVICE_ATTRIBUTES_SHORT_TOPIC:
                        processAttributesSubscribe(grantedQoSList, topicName, reqQoS, TopicType.V2);
                        activityReported = true;
                        break;
                    case MqttTopics.DEVICE_ATTRIBUTES_SHORT_JSON_TOPIC:
                        processAttributesSubscribe(grantedQoSList, topicName, reqQoS, TopicType.V2_JSON);
                        activityReported = true;
                        break;
                    case MqttTopics.DEVICE_ATTRIBUTES_SHORT_PROTO_TOPIC:
                        processAttributesSubscribe(grantedQoSList, topicName, reqQoS, TopicType.V2_PROTO);
                        activityReported = true;
                        break;
                    case MqttTopics.DEVICE_RPC_REQUESTS_SUB_TOPIC:
                        processRpcSubscribe(grantedQoSList, topicName, reqQoS, TopicType.V1);
                        activityReported = true;
                        break;
                    case MqttTopics.DEVICE_RPC_REQUESTS_SUB_SHORT_TOPIC:
                        processRpcSubscribe(grantedQoSList, topicName, reqQoS, TopicType.V2);
                        activityReported = true;
                        break;
                    case MqttTopics.DEVICE_RPC_REQUESTS_SUB_SHORT_JSON_TOPIC:
                        processRpcSubscribe(grantedQoSList, topicName, reqQoS, TopicType.V2_JSON);
                        activityReported = true;
                        break;
                    case MqttTopics.DEVICE_RPC_REQUESTS_SUB_SHORT_PROTO_TOPIC:
                        processRpcSubscribe(grantedQoSList, topicName, reqQoS, TopicType.V2_PROTO);
                        activityReported = true;
                        break;
                    case MqttTopics.DEVICE_RPC_RESPONSE_SUB_TOPIC:
                    case MqttTopics.DEVICE_RPC_RESPONSE_SUB_SHORT_TOPIC:
                    case MqttTopics.DEVICE_RPC_RESPONSE_SUB_SHORT_JSON_TOPIC:
                    case MqttTopics.DEVICE_RPC_RESPONSE_SUB_SHORT_PROTO_TOPIC:
                    case MqttTopics.DEVICE_ATTRIBUTES_RESPONSES_TOPIC:
                    case MqttTopics.DEVICE_ATTRIBUTES_RESPONSES_SHORT_TOPIC:
                    case MqttTopics.DEVICE_ATTRIBUTES_RESPONSES_SHORT_JSON_TOPIC:
                    case MqttTopics.DEVICE_ATTRIBUTES_RESPONSES_SHORT_PROTO_TOPIC:
                    case MqttTopics.GATEWAY_ATTRIBUTES_TOPIC:
                    case MqttTopics.GATEWAY_RPC_TOPIC:
                    case MqttTopics.GATEWAY_ATTRIBUTES_RESPONSE_TOPIC:
                    case MqttTopics.DEVICE_PROVISION_RESPONSE_TOPIC:
                    case MqttTopics.DEVICE_FIRMWARE_RESPONSES_TOPIC:
                    case MqttTopics.DEVICE_FIRMWARE_ERROR_TOPIC:
                    case MqttTopics.DEVICE_SOFTWARE_RESPONSES_TOPIC:
                    case MqttTopics.DEVICE_SOFTWARE_ERROR_TOPIC:
                        registerSubQoS(topicName, grantedQoSList, reqQoS);
                        break;
                    default:
                        log.warn("[{}] not support to subscribe to [{}][{}]", sessionId, topicName, reqQoS);
                        grantedQoSList.add(FAILURE.value());
                        break;
                }
            } catch (Exception e) {
                log.warn("[{}] Failed to subscribe to [{}][{}]", sessionId, topicName, reqQoS, e);
                grantedQoSList.add(FAILURE.value());
            }
        }
        if (!activityReported) {
            transportService.reportActivity(deviceSessionContext.getSessionInfo());
        }
        //订阅应答
        MqttMessageIdVariableHeader messageIdVariableHeader = mqttMsg.variableHeader();
        MqttMessageIdVariableHeader variableHeaderBack = MqttMessageIdVariableHeader.from(messageIdVariableHeader.messageId());
        MqttFixedHeader mqttFixedHeaderBack = new MqttFixedHeader(MqttMessageType.SUBACK, false, MqttQoS.AT_MOST_ONCE, false, 2 + topicSubscriptions.size());
        MqttSubAckPayload payloadBack = new MqttSubAckPayload(grantedQoSList);
        MqttSubAckMessage subAck = new MqttSubAckMessage(mqttFixedHeaderBack, variableHeaderBack, payloadBack);
        log.debug("[{}] Subscription ack: {}!", sessionId, subAck.toString());
        ctx.writeAndFlush(subAck);
    }

    private void processRpcSubscribe(List<Integer> grantedQoSList, String topic, MqttQoS reqQoS, TopicType topicType) {
        //TODO rpc消息订阅处理
        rpcSubTopicType = topicType;
        registerSubQoS(topic, grantedQoSList, reqQoS);
    }

    private void processAttributesSubscribe(List<Integer> grantedQoSList, String topic, MqttQoS reqQoS, TopicType topicType) {
        //TODO 属性消息订阅处理
        this.attrSubTopicType = topicType;
        registerSubQoS(topic, grantedQoSList, reqQoS);
    }

    private void registerSubQoS(String topic, List<Integer> grantedQoSList, MqttQoS reqQoS) {
        grantedQoSList.add(reqQoS.value());
        this.mqttQosMap.putIfAbsent(new MqttTopicMatcher(topic), reqQoS.value());
    }

    /**
     * mqtt凭证认证连接
     *
     * @param ctx
     * @param connectMessage
     */
    private void processBasicCredConnect(ChannelHandlerContext ctx, MqttConnectMessage connectMessage) {

        String userName = connectMessage.payload().userName();
        log.debug("[{}][{}] Processing connect msg for client with user name: {}!", address, sessionId, userName);
        String clientId = connectMessage.payload().clientIdentifier();
        byte[] passwordBytes = connectMessage.payload().passwordInBytes();
        String password = null;
        if (passwordBytes != null) {
            password = new String(passwordBytes, CharsetUtil.UTF_8);
        }

        transportService.basicMqttCredValidate(DeviceTransportType.MQTT, clientId, userName, password,
                new TransportAuthServiceCallback<ValidateDeviceCredentialsResponse>() {
                    @Override
                    public void onSuccess(ValidateDeviceCredentialsResponse msg) {
                        onValidateDeviceResponse(msg, ctx, connectMessage);
                    }

                    @Override
                    public void onError(Throwable e) {
                        log.error("[{}] Failed to process base mqtt credentials: {}", address, userName, e);
                        ctx.writeAndFlush(createMqttConnAckMsg(MqttConnectReturnCode.CONNECTION_REFUSED_SERVER_UNAVAILABLE, connectMessage));
                        ctx.close();
                    }
                });
    }

    /**
     * x509 证书认证连接
     *
     * @param ctx
     * @param cert
     * @param connectMessage
     */
    private void processX509CertConnect(ChannelHandlerContext ctx, X509Certificate cert, MqttConnectMessage connectMessage) {
        try {

            String StrCert = SslUtil.getCertificateString(cert);
            String sha3Hash = SslUtil.getCertificateHash(StrCert);
            transportService.deviceX509CertValidate(DeviceTransportType.MQTT, sha3Hash,
                    new TransportAuthServiceCallback<ValidateDeviceCredentialsResponse>() {
                        @Override
                        public void onSuccess(ValidateDeviceCredentialsResponse msg) {
                            onValidateDeviceResponse(msg, ctx, connectMessage);
                        }

                        @Override
                        public void onError(Throwable e) {
                            log.error("[{}] Failed to process x509 credentials: {}", address, sha3Hash, e);
                            ctx.writeAndFlush(createMqttConnAckMsg(MqttConnectReturnCode.CONNECTION_REFUSED_SERVER_UNAVAILABLE, connectMessage));
                            ctx.close();
                        }
                    });
        } catch (Exception e) {
            ctx.writeAndFlush(createMqttConnAckMsg(CONNECTION_REFUSED_NOT_AUTHORIZED, connectMessage));
            log.trace("[{}] X509 auth failure: {}", sessionId, address, e);
            ctx.close();
        }
    }

    private boolean checkConnected(ChannelHandlerContext ctx, MqttMessage msg) {
        if (deviceSessionContext.isConnected()) {
            return true;
        } else {
            log.info("[{}] Closing current session due to invalid msg order: {}", sessionId, msg);
            return false;
        }
    }

    private void onValidateDeviceResponse(ValidateDeviceCredentialsResponse msg, ChannelHandlerContext ctx, MqttConnectMessage connectMessage) {
        //fail
        if (!msg.hasDeviceInfo()) {
            ctx.writeAndFlush(createMqttConnAckMsg(CONNECTION_REFUSED_NOT_AUTHORIZED, connectMessage));
            ctx.close();
            return;
        }
        this.deviceSessionContext.setClientId(connectMessage.payload().clientIdentifier());
        this.deviceSessionContext.setDeviceInfo(msg.getDeviceInfo());
        this.deviceSessionContext.setDeviceProfile(msg.getDeviceProfile());
        this.deviceSessionContext.setChannel(ctx);
        this.deviceSessionContext.setSessionInfo(SessionInfoCreator.create(msg, this.context, this.sessionId));
        SessionMetaData sessionMetaData = transportService.registerAsyncSession(deviceSessionContext.getSessionInfo(), MqttTransportHandler.this);
        checkGatewaySession(sessionMetaData);
        ctx.writeAndFlush(createMqttConnAckMsg(CONNECTION_ACCEPTED, connectMessage));
        this.deviceSessionContext.setConnected(true);
        log.info("connection success: {} session-id[{}]", deviceSessionContext.getClientId(), this.sessionId);
        this.transportService.getCallbackExecutor().execute(() -> {
            processMsgQueue(ctx);
        });
    }

    private void checkGatewaySession(SessionMetaData sessionMetaData) {

        TransportDeviceInfo device = deviceSessionContext.getDeviceInfo();
        try {
            JsonNode infoNode = context.getMapper().readTree(device.getAdditionalInfo());
            if (infoNode != null) {
                JsonNode gatewayNode = infoNode.get(DeviceAdditionalConstants.GATEWAY);
                if (gatewayNode != null && gatewayNode.asBoolean()) {
                    gatewaySessionHandler = new GatewaySessionHandler(deviceSessionContext, sessionId);
                    if (infoNode.has(DeviceAdditionalConstants.OVERWRITE_ACTIVITY_TIME) && infoNode.get(DeviceAdditionalConstants.OVERWRITE_ACTIVITY_TIME).isBoolean()) {
                        sessionMetaData.setOverwriteActivityTime(infoNode.get(DeviceAdditionalConstants.OVERWRITE_ACTIVITY_TIME).asBoolean());
                    }
                }
            }
        } catch (IOException e) {
            log.trace("[{}][{}] Failed to fetch device additional info", sessionId, device.getDeviceName(), e);
        }
    }

    private X509Certificate getX509Certificate() {
        try {
            Certificate[] certChain = sslHandler.engine().getSession().getPeerCertificates();
            if (certChain.length > 0) {
                return (X509Certificate) certChain[0];
            }
        } catch (SSLPeerUnverifiedException e) {
            log.warn(e.getMessage());
            return null;
        }
        return null;
    }

    private MqttConnAckMessage createMqttConnAckMsg(MqttConnectReturnCode returnCode, MqttConnectMessage msg) {
        MqttFixedHeader mqttFixedHeader =
                new MqttFixedHeader(CONNACK, false, AT_MOST_ONCE, false, 0);
        MqttConnAckVariableHeader mqttConnAckVariableHeader =
                new MqttConnAckVariableHeader(returnCode, !msg.variableHeader().isCleanSession());
        return new MqttConnAckMessage(mqttFixedHeader, mqttConnAckVariableHeader);
    }

    void processMsgQueue(ChannelHandlerContext ctx) {
        if (!deviceSessionContext.isConnected()) {
            log.trace("[{}][{}] Postpone processing msg due to device is not connected. Msg queue size is {}", sessionId, deviceSessionContext.getDeviceId(), deviceSessionContext.getMsgQueueSize());
            return;
        }
        this.deviceSessionContext.tryProcessQueueMessage(msg -> processRegularSessionMsg(ctx, msg));
    }

    private InetSocketAddress getAddress(ChannelHandlerContext ctx) {

        InetSocketAddress address = ctx.channel().attr(BootNettyMqttServer.ADDRESS).get();
        if (address == null) {
            log.trace("[{}] Received empty address.", ctx.channel().id());
            InetSocketAddress remoteAddress = (InetSocketAddress) ctx.channel().remoteAddress();
            log.trace("[{}] Going to use address: {}", ctx.channel().id(), remoteAddress);
            return remoteAddress;
        } else {
            log.trace("[{}] Received address: {}", ctx.channel().id(), address);
        }
        return address;
    }

    private <T> TransportServiceCallback<Void> getPubAckCallback(final ChannelHandlerContext ctx, final int msgId, final MqttFixedHeader mqttFixedHeader,
                                                                 final T msg) {
        return new TransportServiceCallback<>() {
            @Override
            public void onSuccess(Void dummy) {
                log.trace("[{}] Published msg: {}", sessionId, msg);
                publishAck(ctx, msgId, mqttFixedHeader);
            }

            @Override
            public void onError(Throwable e) {
                log.trace("[{}] Failed to publish msg: {}", sessionId, msg, e);
                ctx.close();
            }
        };
    }

    private class DeviceProvisionCallback implements TransportServiceCallback<TransportProto.ProvisionDeviceResponseMsg> {

        private final ChannelHandlerContext ctx;
        private final int msgId;
        private MqttFixedHeader mqttFixedHeader;
        private final TransportProto.ProvisionDeviceRequestMsg msg;

        DeviceProvisionCallback(ChannelHandlerContext ctx, int msgId, MqttFixedHeader mqttFixedHeader, TransportProto.ProvisionDeviceRequestMsg msg) {
            this.ctx = ctx;
            this.msgId = msgId;
            this.mqttFixedHeader = mqttFixedHeader;
            this.msg = msg;
        }

        @Override
        public void onSuccess(TransportProto.ProvisionDeviceResponseMsg provisionResponseMsg) {
            log.trace("[{}] Published msg: {}", sessionId, msg);
            publishAck(this.ctx, this.msgId, this.mqttFixedHeader);
            try {
                switch (deviceSessionContext.getProvisionPayloadType()) {
                    case JSON:
                        MqttTransportAdaptorFactory.getJsonMqttAdaptor().convertToPublish(deviceSessionContext, provisionResponseMsg)
                                .ifPresent(deviceSessionContext.getChannel()::writeAndFlush);
                        break;
                    case PROTOBUF:
                        MqttTransportAdaptorFactory.getProtoMqttAdaptor().convertToPublish(deviceSessionContext, provisionResponseMsg)
                                .ifPresent(deviceSessionContext.getChannel()::writeAndFlush);
                        break;
                }
                scheduler.schedule((Callable<ChannelFuture>) ctx::close, 60, TimeUnit.SECONDS);
            } catch (Exception e) {
                log.trace("[{}] Failed to convert device attributes response to MQTT msg", sessionId, e);
            }
        }

        @Override
        public void onError(Throwable e) {
            log.trace("[{}] Failed to publish msg: {}", sessionId, msg, e);
            ctx.close();
        }
    }

    @Override
    public void onRemoteSessionCloseCommand(UUID sessionId, TransportProto.SessionCloseNotificationProto sessionCloseNotification) {
        log.trace("[{}] Received the remote command to close the session: {}", sessionId, sessionCloseNotification.getMessage());
        deviceSessionContext.getChannel().close();
    }

    @Override
    public void onGetAttributesResponse(TransportProto.GetAttributeResponseMsg attributesResponse) {

        log.trace("[{}] Received get attributes response", sessionId);
        String topicBase = attrReqTopicType.getAttributesResponseTopicBase();
        MqttTransportAdaptor adaptor = deviceSessionContext.getAdaptor(attrReqTopicType);
        try {
            Optional<MqttMessage> mqttMessage = adaptor.convertToPublish(deviceSessionContext, attributesResponse, topicBase);
            mqttMessage.ifPresent(message -> deviceSessionContext.getChannel().writeAndFlush(message));
        } catch (AdaptorException e) {
            log.trace("[{}] Failed to convert device attributes response to MQTT msg", sessionId, e);
        }
    }

    @Override
    public void onAttributeUpdate(UUID sessionId, TransportProto.AttributeUpdateNotificationMsg attributeUpdateNotification) {

        log.trace("[{}] Received attributes update notification to device", sessionId);
        String topic = attrSubTopicType.getAttributesSubTopic();
        MqttTransportAdaptor adaptor = deviceSessionContext.getAdaptor(attrSubTopicType);
        try {
            Optional<MqttMessage> mqttMessage = adaptor.convertToPublish(deviceSessionContext, attributeUpdateNotification, topic);
            if (mqttMessage.isPresent()) {
                deviceSessionContext.getChannel().writeAndFlush(mqttMessage);
            }
        } catch (AdaptorException e) {
            log.trace("[{}] Failed to convert device attributes update to MQTT msg", sessionId, e);
        }
    }

    @Override
    public void onDeviceUpdate(SessionProto.SessionInfoProto sessionInfo, Device device, Optional<DeviceProfile> deviceProfileOpt) {

    }

    @Override
    public void onDeviceDeleted(Long deviceId) {

    }

    @Override
    public void onDeviceProfileUpdate(SessionProto.SessionInfoProto newSessionInfo, DeviceProfile deviceProfile) {

    }

    @Override
    public void onToDeviceRpcRequest(UUID sessionId, TransportProto.ToDeviceRpcRequestMsg toDeviceRequest) {

    }

    @Override
    public void onToServerRpcResponse(TransportProto.ToServerRpcResponseMsg toServerResponse) {

    }
}
